package main

import (
	"fmt"
	"github.com/segmentio/kafka-go"
	"context"
	"log"
)

func main() {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:   []string{"10.111.55.75:9092","10.111.55.73:9092","10.111.55.74:9092"},
		Topic:     "svc-count-flink-prod",
		Partition: 0, //这个不起作用，实际消费时还是消费所有的分区
		MinBytes:  10e3, // 10KB
		MaxBytes:  10e6, // 10MB
		GroupID: "prometheus_test",
		CommitInterval: 3*time.Second,
	})
	//r.SetOffset(42)

	for {
		m, err := r.ReadMessage(context.Background())
		if err != nil {
			fmt.Println(11111, err)
			break
		}
		//fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
		fmt.Println(11111, m.Offset, m.Partition)
		fmt.Println(2222, string(m.Value))
                break
	}

	if err := r.Close(); err != nil {
		log.Fatal("failed to close reader:", err)
	}
}
